// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

// Slow stream implementations, mainly for testing and benchmarking

#pragma once

#include <cstdint>
#include <memory>
#include <utility>

#include <nebula/io/interfaces.h>
#include <turbo/base/macros.h>

namespace nebula {

    class Buffer;


    namespace io {

        class TURBO_EXPORT LatencyGenerator {
        public:
            virtual ~LatencyGenerator();

            void Sleep();

            virtual double NextLatency() = 0;

            static std::shared_ptr<LatencyGenerator> create(double average_latency);

            static std::shared_ptr<LatencyGenerator> create(double average_latency, int32_t seed);
        };

        // XXX use ConcurrencyWrapper?  It could increase chances of finding a race.

        template<class StreamType>
        class SlowInputStreamBase : public StreamType {
        public:
            SlowInputStreamBase(std::shared_ptr<StreamType> stream,
                                std::shared_ptr<LatencyGenerator> latencies)
                    : stream_(std::move(stream)), latencies_(std::move(latencies)) {}

            SlowInputStreamBase(std::shared_ptr<StreamType> stream, double average_latency)
                    : stream_(std::move(stream)), latencies_(LatencyGenerator::create(average_latency)) {}

            SlowInputStreamBase(std::shared_ptr<StreamType> stream, double average_latency,
                                int32_t seed)
                    : stream_(std::move(stream)),
                      latencies_(LatencyGenerator::create(average_latency, seed)) {}

        protected:
            std::shared_ptr<StreamType> stream_;
            std::shared_ptr<LatencyGenerator> latencies_;
        };

        /// \brief An InputStream wrapper that makes reads slower.
        ///
        /// read() calls are made slower by an average latency (in seconds).
        /// Actual latencies form a normal distribution closely centered
        /// on the average latency.
        /// Other calls are forwarded directly.
        class TURBO_EXPORT SlowInputStream : public SlowInputStreamBase<InputStream> {
        public:
            ~SlowInputStream() override;

            using SlowInputStreamBase<InputStream>::SlowInputStreamBase;

            turbo::Status close() override;

            turbo::Status abort() override;

            bool closed() const override;

            turbo::Result<int64_t> read(int64_t nbytes, void *out) override;

            turbo::Result<std::shared_ptr<Buffer>> read(int64_t nbytes) override;

            turbo::Result<std::string_view> peek(int64_t nbytes) override;

            turbo::Result<int64_t> tell() const override;
        };

        /// \brief A RandomAccessFile wrapper that makes reads slower.
        ///
        /// Similar to SlowInputStream, but allows random access and seeking.
        class TURBO_EXPORT SlowRandomAccessFile : public SlowInputStreamBase<RandomAccessFile> {
        public:
            ~SlowRandomAccessFile() override;

            using SlowInputStreamBase<RandomAccessFile>::SlowInputStreamBase;

            turbo::Status close() override;

            turbo::Status abort() override;

            bool closed() const override;

            turbo::Result<int64_t> read(int64_t nbytes, void *out) override;

            turbo::Result<std::shared_ptr<Buffer>> read(int64_t nbytes) override;

            turbo::Result<int64_t> read_at(int64_t position, int64_t nbytes, void *out) override;

            turbo::Result<std::shared_ptr<Buffer>> read_at(int64_t position, int64_t nbytes) override;

            turbo::Result<std::string_view> peek(int64_t nbytes) override;

            turbo::Result<int64_t> get_size() override;

            turbo::Status Seek(int64_t position) override;

            turbo::Result<int64_t> tell() const override;
        };

    }  // namespace io
}  // namespace nebula
